package mc.support.service.common.impl;

import mc.support.service.common.ICacheService;
import mc.support.util.SerializeUtil;
import com.taobao.api.domain.Item;
import org.apache.commons.lang.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import redis.clients.jedis.Jedis;
import redis.clients.jedis.JedisPool;

import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;

/**
 * Created with IntelliJ IDEA. <br/>
 * Author: francis Yun    <br/>
 * Date: 2013-11-23  <br/>
 *
 *
 * exception of cache service shall not affect business logic!!!
 */
public class CacheServiceImpl implements ICacheService {

    private static final Logger log = LoggerFactory.getLogger( CacheServiceImpl.class );

    private static final int DEFAULT_EXPIRE_SEC = 60 * 30;

    private CacheKeyMgr cacheKeyMgr = new CacheKeyMgr();

    private JedisPool jedisPool;

    private Map<String, Set<String>> keyToDelSet = new ConcurrentHashMap<String, Set<String>>();

    public void setJedisPool(JedisPool jedisPool) {
        this.jedisPool = jedisPool;
    }

    @Override
    public boolean writeObjectToCache(Object obj, String key) {
        return writeObjectToCache( obj, key, DEFAULT_EXPIRE_SEC );
    }

    @Override
    public boolean writeObjectToCache(Object obj, String key, int expireSec) {
        if ( null == obj ) {
            return false;
        }

        boolean success = true;
        Jedis jedis = null;
        try {
            jedis = getJedis();
            if ( null != jedis ) {
                byte[] bs = SerializeUtil.serializeToByteArray(obj);

                if ( StringUtils.isNotBlank( key ) && null != bs ) {
                    jedis.set(key.getBytes(), bs);
                    jedis.expire(key.getBytes(), expireSec);
                }
            }
        } catch (Exception e ) {
            log.error( "jedis: write cache error: " + e.getMessage() );
            success = false;
        } finally {
            returnJedis( jedis );
        }

        return success;
    }

    @Override
    public boolean writeItemsToCache(List<Item> items, String key) {
        return writeItemsToCache( items, key, DEFAULT_EXPIRE_SEC );
    }

    @Override
    public boolean writeItemsToCache(List<Item> items, String key, int expireSec) {
        if ( null == items || 0 == items.size() ) {
            return false;
        }
        return writeObjectToCache( items, key, expireSec );
    }

    @Override
    public void registerKeyToDel(String subject, String key) {
        Set<String> keySet = keyToDelSet.get( subject );
        if ( null == keySet ) {
            synchronized (keyToDelSet) {
                keySet = keyToDelSet.get( subject );
                if ( null == keySet ) {
                    keySet = new HashSet<String>();
                    keyToDelSet.put(subject, keySet);
                }
            }
        }
        synchronized ( keySet ) {
            keySet.add( key );
        }
    }

    @Override
    public boolean delKeysBySubject(String subject) {
        Set<String> keySet = keyToDelSet.get( subject );
        boolean success = true;
        if ( null != keySet ) {
            synchronized ( keySet ) {
                try {
                    for ( String key : keySet ) {
                        success = success && delKey(key);
                    }
                } finally {
                    keySet.clear();
                }
            }
        }
        return success;
    }

    @Override
    public boolean delKey(String key) {
        boolean success = true;
        Jedis jedis = null;
        try {
            jedis = getJedis();
            if ( null != jedis ) {
                jedis.del( key.getBytes() );
            }
        } catch (Exception e) {
            log.error( "jedis: delete key error: " + e.getMessage() );
            success = false;
        } finally {
            returnJedis( jedis );
        }
        return success;
    }

    @Override
    public Object readObjectFromCache(String key) {
        if ( !StringUtils.isNotBlank(key) ) {
            return  null;
        }

        Object obj = null;
        Jedis jedis = null;
        try {
            jedis = getJedis();
            byte[] bs = jedis.get( key.getBytes() );
            if ( null != bs && bs.length > 0 ) {
                obj = SerializeUtil.deserializeToObject( bs );
            }
        } catch (Exception e) {
            log.error( "jedis: read cache error: " + e.getMessage() );
        } finally {
            returnJedis( jedis );
        }
        return obj;
    }

    @Override
    public List<Item> readItemsFromCache(String key) {
        Object cached = readObjectFromCache( key );
        List<Item> items = null;
        if ( null != cached && cached instanceof List ) {
            //still not type safe
            items = (List<Item>) cached;
        }

        return items;
    }

    @Override
    public String generateCacheKey(Class<?> clazz, String... subjects) {
        return cacheKeyMgr.getKey( clazz, subjects );
    }

    @Override
    public boolean isCached(String cachedKey) {

        if ( !StringUtils.isNotBlank( cachedKey ) ) {
            return false;
        }

        boolean isCached = false;
        Jedis jedis = null;
        try {
            jedis = getJedis();
            isCached = jedis.exists( cachedKey.getBytes() );
        } catch (Exception e) {
            log.error( "jedis error: " + e.getMessage() );
        } finally {
            returnJedis( jedis );
        }

        return isCached;
    }

    private Jedis getJedis() {
        Jedis jedis = null;
        try {
            jedis = jedisPool.getResource();
        } catch (Exception e) {
            log.error( "jedis error: " + e.getMessage() );
        }

        return jedis;
    }

    private void returnJedis(Jedis jedis) {
        if ( null == jedis ) {
            return;
        }
        try {
            jedisPool.returnResource( jedis );
        } catch (Exception e) {
            log.error( "return jedis error" + e.getMessage() );
        }
    }

}
